package industry_2024.industry_04.tzgc


import org.apache.spark.sql.{Dataset, Row, SparkSession}
import org.dom4j.DocumentHelper

import java.sql.Timestamp
import java.text.SimpleDateFormat
import java.util.Properties


object tzgc01 {
  def main(args: Array[String]): Unit = {
    /*
        1、根据dwd库中fact_machine_data表（或MySQL的shtd_industry库中MachineData表），根据以下要求转换：获取最大分
        区（MySQL不用考虑）的数据后，首先解析列machine_record_data（MySQL中为MachineRecordData）的数据（数据格式
        为xml，采用dom4j解析，解析demo在客户端/home/ubuntu/Documents目录下），并获取每条数据的主轴转速，主轴倍率，主轴
        负载，进给倍率，进给速度，PMC程序号，循环时间，运行时间，有效轴数，总加工个数，已使用内存，未使用内存，可用程序量，注
        册程序量等相关的值（若该条数据没有相关值，则按下表设置默认值），同时转换machine_record_state字段的值，若值为报
        警，则填写1，否则填写0，以下为表结构，将数据保存在dwd.fact_machine_learning_data，使用cli按
        照machine_record_id升序排序，查询dwd.fact_machine_learning_data前1条数据，将结果截图粘贴至客户端桌
        面【Release\任务C提交结果.docx】中对应的任务序号下。
    dwd.fact_machine_learning_data表结构：
     */

    val spark=SparkSession.builder()
      .master("local[*]")
      .appName("第四套卷子的特征工程第一题")
      .config("hive.exec.dynamic.partition.mode","nonstrict")
      .config("spark.serializer","org.apache.spark.serializer.KryoSerializer")
      .config("spark.sql.extensions","org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
      .enableHiveSupport()
      .getOrCreate()

    //  为了避免多余的报错导入spark的隐式转换
    import spark.implicits._

    val connect=new Properties()
    connect.setProperty("user","root")
    connect.setProperty("password","123456")
    connect.setProperty("driver","com.mysql.jdbc.Driver")


    val machine_data=spark.read
      .jdbc("jdbc:mysql://192.168.40.110:3306/shtd_industry?useSSL=false","MachineData",connect)

//    println(machine_data.columns.mkString(","))

    //  将字段的名字更改成小写
    val data=machine_data
      .withColumnRenamed("MachineRecordID","machine_record_id")
      .withColumnRenamed("MachineID","machine_id")
      .withColumnRenamed("MachineRecordState","machine_record_state")
      .withColumnRenamed("MachineRecordDate","machine_record_date")

    println("---------------更改字段名称后的结果:---------------")
    println(data.columns.mkString(","))


    val machineList:Dataset[machine]=data.map(
      row =>{
        //  创建一个machine对象，并且对其一些属性进行赋值
        //  row.getAs 是 Scala 中用于从某种数据结构（通常是一个行数据结构，比如 Spark 的Row`）中获取特定列的值的方法。
        //  getAs[Int](0):int是你希望将获得的数据转换为什么数据类型，这里的话，将每一行的第一列数据转化为int类型
        val m=new machine()
        m.machine_record_id=row.getAs[Int](0)
        m.machine_id=row.getAs[Int](1)
        m.machine_record_state=if (row.getAs[String](2).equals("报警")) 1.0 else 0.0
        /*
            new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(...)：format 方法是用于将 Date 对象转换为字符串，而你传入的是一个字符串，这样
            会导致不正确的结果或抛出异常。,所以下面必须使用parse(解析)将字符串转化为Date,然后使用getTime方法将Date转化为时间戳
         */
        m.machine_record_date=row.getAs[Timestamp](4)

        //  如果MachineRecordData字段不是空值的话就进入下面的操作
        //  处理从某个数据行 (row) 中获取的 XML 数据，进行解析并遍历其子元素。
        if(row.getAs[String](3)!=null){
          //  1.构建完整的xml字符串,这里的<rows></rows>标签是必须要加的，而不是为了整齐美观
          val machineData:String=s"<rows>${row.getAs[String](3)}</rows>"
          //  2.使用DocumentHelper.parseText方法解析xml字符串，将其转化为一个document对象，以便后面的操作
          //  Helper:助手
          val document=DocumentHelper.parseText(machineData)
          //  3.从解析后的文档中获取根元素，这里应该是<rows>标签
          val root=document.getRootElement
          //  4.获取根元素的子元素的迭代器，准备遍历他们
          val it=root.elementIterator()
          //  5.开始遍历元素
          while(it.hasNext){
            //  6.从迭代器中获取下一个子元素。
            val element=it.next()
            //  获取当前元素的 ColName 属性值。这通常用于标识元素的含义或相关数据。 attribute:属性
            val colName=element.attributeValue("ColName")
            //  检查当前元素的文本是否不为 null 且不为空
            //  getTextTrim会获取元素的文本内容，并且会去除前后的空格  trim:修剪
            if(!element.getTextTrim.equals("null") && element.getTextTrim.nonEmpty){
                  colName match{
                    case "主轴转速" => m.machine_record_mainshaft_speed=element.getTextTrim.toDouble
                    case "主轴倍率" => m.machine_record_mainshaft_multiplerate=element.getTextTrim.toDouble
                    case "主轴负载" => m.machine_record_mainshaft_load=element.getTextTrim.toDouble
                    case "进给倍率" => m.machine_record_feed_speed=element.getTextTrim.toDouble
                    case "进给速度" =>  m.machine_record_feed_multiplerate=element.getTextTrim.toDouble
                    case "PMC程序号" => m.machine_record_pmc_code=element.getTextTrim.toDouble
                    case "循环时间" => m.machine_record_circle_time=element.getTextTrim.toDouble
                    case "运行时间" => m.machine_record_run_time=element.getTextTrim.toDouble
                    case "有效轴数" => m.machine_record_effective_shaft=element.getTextTrim.toDouble
                    case "总加工个数" => m.machine_record_amount_process=element.getTextTrim.toDouble
                    case "已使用内存" => m.machine_record_use_memory=element.getTextTrim.toDouble
                    case "未使用内存" => m.machine_record_free_memory=element.getTextTrim.toDouble
                    case "可用程序量" => m.machine_record_amount_use_code=element.getTextTrim.toDouble
                    case "注册程序量" => m.machine_record_amount_free_code=element.getTextTrim.toDouble
                    case _ => ""
                  }
            }
          }
        }
        //  将结果返回
        m
      }
    )
  //  查看结果
    machineList.limit(20).show()

    machineList.write.mode("overwrite").saveAsTable("xiaoqiu.fact_machine_learning_data")
    println("写入完成")

    spark.close()
  }
}


//  下面定义一个样例类
case class machine(
                  //  System.currentTimeMillis():返回当前的时间戳
                  //  new Timestamp():将时间转化为timestamp()数据类型
                  //  要注意的是下面的Timestamp是在这个包里面的，别的会报错import java.sql.Timestamp
                  var machine_record_id:Int=0,
                  var machine_id:Double=0.0,
                  var machine_record_state:Double=0.0,
                  var machine_record_mainshaft_speed:Double=0.0,            //  mainshaft:主轴   speed:速度
                  var machine_record_mainshaft_multiplerate:Double=0.0,      // multiplerate:倍率
                  var machine_record_mainshaft_load:Double=0.0,               //  主轴负载
                  var machine_record_feed_speed:Double=0.0,                   //  进给倍率
                  var machine_record_feed_multiplerate:Double=0.0,            //  进给速度
                  var machine_record_pmc_code:Double=0.0,                     //  pmc程序号
                  var machine_record_circle_time:Double=0.0,                  //  circle:圆      循环时间
                  var machine_record_run_time:Double=0.0,                     //  运行时间
                  var machine_record_effective_shaft:Double=0.0,              //  有效轴数      effective:有效的
                  var machine_record_amount_process:Double=0.0,               //  总加工个数
                  var machine_record_use_memory:Double=0.0,                   //  已使用内存
                  var machine_record_free_memory:Double=0.0,                  //  未使用的内存
                  var machine_record_amount_use_code:Double=0.0,
                  var machine_record_amount_free_code:Double=0.0,
                  var machine_record_date:Timestamp= new Timestamp(System.currentTimeMillis()),      //  import java.sql.Timestamp
                  var dwd_insert_user:String="user1",
                  var dwd_insert_time:Timestamp=new Timestamp(System.currentTimeMillis()),          //  import java.sql.Timestamp
                  var dwd_modify_user:String="user1",
                  var dwd_modify_time:Timestamp=new Timestamp(System.currentTimeMillis())           //  import java.sql.Timestamp

                  )
